package com.peng.sparktest.sparkcore

import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

object SparkApiTest {
  def main(args: Array[String]): Unit = {

    val conf: SparkConf = new SparkConf().setMaster("local").setAppName("apiTest")
    val context = new SparkContext(conf)
    context.setLogLevel("ERROR")
    //    //从内存中构建一个RDD
    //        val data: RDD[Int] = context.parallelize(List(1, 2, 3, 4, 5, 3, 1, 2))
    //    data.foreach(println)
    //    val filterData: RDD[Int] = data.filter(_ > 3)
    //    val list: Array[Int] = filterData.collect()
    //    list.foreach(println) //本地数据集的foreach
    //    filterData.foreach(println)
    //
    //    data.distinct().foreach(println)


    //    val data: RDD[Int] = context.parallelize(List(1, 2, 3, 4, 5))
    //    val data2: RDD[Int] = context.parallelize(List(4, 5, 6, 7))


    //并集(实际不存放数据，内部每一个位置指向真实RDD分区的信息)
    //    println("==========================并集============================")
    //        val unionData: RDD[Int] = data.union(data2)
    //
    //    unionData.foreach(println)

    //笛卡尔积（未发生shuffle操作，单纯的使用IO操作将data2的数据全量拉了过来）

    // 不同机器RDD数据需要汇总计算时的两种数据合并策略：
    //1、shuffle拉取   一边shuffle写，一边shuffle读
    //2、直接IO拉取，不走分区计算逻辑，直接将数据全量拉取过来

    //如果 数据加工逻辑不需要区分每一条记录的分区，也就是这样的数据不需要分区器，也不要shuffle，那这样的话，
    //直接IO拉取，比先使用分区器计算，然后shuffle落地文件，再IO拉取效率要高
    //    println("==========================笛卡尔积============================")
    //    val cartRes: RDD[(Int, Int)] = data.cartesian(data2)
    //
    //    cartRes.foreach(println)

    //    println("==========================并集============================")
    //    data.intersection(data2).foreach(println)
    //    println("==========================差集============================")
    //    data.subtract(data2).foreach(println)


    //
    val data1: RDD[(String, Int)] = context.parallelize(List(
      ("xiaoming", 2),
      ("xiaoming", 3),
      ("xiaohong", 4)
    ))
    val data2: RDD[(String, Int)] = context.parallelize(List(
      ("xiaoming", 4),
      ("xiaoming", 6),
      ("xiaoliu", 4)
    ))

    println("==========================cogroup============================")
    data1.cogroup(data2).foreach(println)
    //        println("==========================join============================")
    data1.join(data2).foreach(println)
    //        println("==========================leftOuterJoin============================")
    //        data1.leftOuterJoin(data2).foreach(println)
    //        println("==========================rightOuterJoin============================")
    //        data1.rightOuterJoin(data2).foreach(println)
    //        println("==========================fullOuterJoin============================")
    //        data1.fullOuterJoin(data2).foreach(println)

    while (true) {

    }


  }

}
